package com.apes.framework.plugin.outbox;

import com.alibaba.fastjson.JSONObject;
import com.apes.framework.api.DomainService;
import com.apes.framework.jpa.ApesBean;
import com.apes.framework.jpa.BaseModel;
import com.apes.framework.jpa.repository.CustomRepository;
import com.apes.framework.jpa.spec.JpaDsl;
import com.apes.framework.plugin.esb.api.SimpleRequest;
import com.apes.framework.plugin.event.Listener;
import com.apes.framework.plugin.outbox.annotation.OutBoxExclude;
import com.apes.framework.plugin.soe.core.context.Command;
import com.apes.framework.plugin.soe.core.context.SoeContext;
import com.apes.framework.rop.session.SimpleSession;
import com.apes.framework.util.*;
import com.apes.oap.session.RopSessionHolder;
import com.apes.oap.session.Session;
import org.dozer.Mapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Sort;
import org.springframework.data.jpa.domain.Specification;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Service;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.StringUtils;

import java.io.Serializable;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;
import java.util.stream.Collectors;

import static com.apes.framework.util.ReflectUtil.getMerge;

@Service("outboxEventService")
public class OutboxEventService extends DomainService {

    @Value("${apes.outbox.async:true}")
    private boolean isAsync;

    @Value("${apes.outbox.backup:true}")
    private boolean isBackup;

    @Autowired
    private OutboxEventRepository outboxEventRepository;

    @Autowired
    private OutboxEventBackupRepository outboxEventBackupRepository;

    @Autowired
    private OutboxSubscriberManager outboxSubscriberManager;

    @Autowired
    private TransactionTemplate transactionTemplate;

    @Autowired
    private DistributedLock distributedLock;

    public <S> void sendSaveEvent(S entity) {
        this.save("S", entity);
    }

    public <S> void sendDeleteEvent(S entity) {
        this.save("D", entity);
    }

    /**上线临时使用**/
    public <S> void sendSaveEvent1(S entity) {
        this.save("S", entity);
    }

    private <S> void save(String eventType, S entity) {
        if (filter(entity)) return;
        List<OutboxSubscriber> subscribers = getOutboxSubscribers(entity, eventType);
        if (subscribers == null || subscribers.isEmpty()) {
            return;
        }
        publish(entity, subscribers);
    }

    private <S> boolean filter(S entity) {
        return ReflectUtil.getAnnotation(entity.getClass(), OutBoxExclude.class) != null;
    }

    private <S> List<OutboxSubscriber> getOutboxSubscribers(S entity, String eventType) {
        String aggregateType = getAggregateType(entity);
        List<OutboxSubscriber> subscribers = getOutboxSubscribers(eventType, aggregateType);
        if (entity instanceof OutboxPayloadWrapper) {
            OutboxPayloadWrapper wrapper = (OutboxPayloadWrapper) entity;
            subscribers = subscribers.stream().filter(subscriber -> wrapper.getSubscriber().equals(subscriber.getId())).collect(Collectors.toList());
        }
        subscribers = subscribers.stream().distinct().collect(Collectors.toList());
        return subscribers;
    }

    /**
     * 消息分发
     **/
    private <S> void publish(S entity, List<OutboxSubscriber> subscribers) {
        String payload;
        String aggregateId = null;
        if (entity instanceof ApesBean) {
            payload = JSONObject.toJSONString(MapUtil.mapped(entity));
            aggregateId = JpaUtil.getIdToStr((ApesBean) entity);
        } else {
            payload = JSONObject.toJSONString(entity);
            if (entity instanceof OutboxPayloadWrapper) {
                aggregateId = ((OutboxPayloadWrapper) entity).getAggregateId();
            }
        }
        String aggregateType = getAggregateType(entity);
        Map merge = getMerge(entity);
        for (OutboxSubscriber subscriber : subscribers) {
            if (!eval(subscriber.getCondition(), merge)) {
                continue;
            }

            if (isAsync) {
                OutboxEvent outboxEvent = new OutboxEvent();
                outboxEvent.setAggregateId(aggregateId);
                outboxEvent.setAggregateType(aggregateType);
                outboxEvent.setEventType(subscriber.getEventType());
                outboxEvent.setTimestamp(LocalDateTime.now().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
                outboxEvent.setPayload(payload);
                outboxEvent.setSubscriber(subscriber.getId());
                Session session = RopSessionHolder.get();
                if (session != null && session instanceof SimpleSession) {
                    outboxEvent.setSessionId(((SimpleSession) session).getSessionId());
                }
                outboxEvent = outboxEventRepository.saveAndFlush(outboxEvent);
                pushMQ(outboxEvent);
            } else {
                if ("soe.consume".equals(subscriber.getConsumer())) {
                    Command command = new Command(entity);
                    SoeContext.add(command);
                } else if (StringUtils.hasText(subscriber.getConsumer())) {
                    this.invoke(subscriber.getConsumer(), entity);
                } else if (subscriber.getProcessor() != null) {
                    subscriber.getProcessor().apply(entity);
                }
            }
        }
    }

    /**
     * 获取业务模型类型
     **/
    private <S> String getAggregateType(S entity) {
        String aggregateType;
        if (entity instanceof ApesBean) {
            aggregateType = JpaUtil.getDomainClass((ApesBean) entity).getName();
        } else {
            aggregateType = entity.getClass().getName();
        }
        return aggregateType;
    }

    /**
     * 执行表达式，有错误不抛异常，给默认值
     **/
    private boolean eval(final String express, final Map<String, Object> context) {
        if (StringUtils.isEmpty(express)) {
            return true;
        }
        try {
            return (Boolean) CacheUtil.getEngine().execute(express, context);
        } catch (Exception e) {
            return false;
        }
    }

    /**
     * 消息推送MQ
     */
    private void pushMQ(OutboxEvent outboxEvent) {
        Object body = MapUtil.mapper("id", outboxEvent.getId(), "subscriber", outboxEvent.getSubscriber(), "storeTime", System.currentTimeMillis());
        sendAynEvent("event:outbox.message.publish", body);
    }

    public List<OutboxSubscriber> getOutboxSubscribers(String eventType, String aggregateType) {
        List<OutboxSubscriber> subscribers = outboxSubscriberManager.getOutboxSubscribers(eventType, aggregateType);
        List<OutboxSubscriber> outboxSubscribers = outboxSubscriberManager.getOutboxSubscribers(eventType, "*");
        if (subscribers == null) {
            subscribers = outboxSubscribers;
        } else if (outboxSubscribers != null) {
            subscribers.addAll(outboxSubscribers);
        }
        return subscribers;
    }

    @Listener(topic = "event:outbox.message.consume")
    public void consume(SimpleRequest request) {
        Map body = request.getMap();
        String id = (String) body.get("id");
        long storeTime = (Long) body.get("storeTime");
        distributedLock.lock("outbox_" + id, 120 * 1000, () -> {
            OutboxEvent event = findOutboxEvent(id, storeTime);
            if (event == null) return;
            try {
                handlingEvent(event);
            } catch (Exception e) {
                e.printStackTrace();
                handlingException(event, e);
            }
            handlingNextEvent(event); /**优化后置消息的消费速度**/
        }, () -> {
            throw new RuntimeException("消息处理中，请稍等！");
        });
    }

    /**
     * 查找事件消息
     **/
    private OutboxEvent findOutboxEvent(String id, Long storeTime) {
        OutboxEvent event = outboxEventRepository.findOne(id);
        if (event != null) return event;
        int expire = 2; /**单位：分钟**/
        if (DateUtil.add(new Date(), Calendar.MINUTE, -1 * expire).getTime() > storeTime) {
            return null; /**超出时间，则认为收件箱消息已回滚**/
        }
        throw new RuntimeException("收件箱未找到消息，请稍等！");
    }

    /**
     * 消费事件消息
     **/
    private void handlingEvent(OutboxEvent event) {
        execute(() -> {
            checkAdvance(event);
            OutboxSubscriber outboxSubscriber = getOutboxSubscriber(event);
            Object payload = getPayload(event);
            /**处理事件消息**/
            if (StringUtils.hasText(outboxSubscriber.getConsumer())) {
                invoke(outboxSubscriber.getConsumer(), payload);
            } else if (outboxSubscriber.getProcessor() != null) {
                outboxSubscriber.getProcessor().apply(payload);
            } else {
                throw new RuntimeException("收件箱消息消费失败！");
            }
            /**备份事件消息**/
            if (isBackup) {
                OutboxEventBackup outboxEventBackup = new OutboxEventBackup();
                ReflectUtil.copyAll(event, outboxEventBackup, "id");
                outboxEventBackup.setOutboxEventId(event.getId());
                outboxEventBackup.setTimestamp(LocalDateTime.now().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
                outboxEventBackupRepository.save(outboxEventBackup);
            }
            /**删除事件消息**/
            outboxEventRepository.delete(event);
        });
    }

    /**
     * 检查是否提前消费
     **/
    private void checkAdvance(OutboxEvent event) {
        int count = outboxEventRepository.countAllBySubscriberAndAggregateTypeAndAggregateIdAndIdLessThan(
                event.getSubscriber(), event.getAggregateType(), event.getAggregateId(), event.getId());
        if (count > 0) {
            throw new RuntimeException("收件箱中还存在前置消息！");
        }
    }

    /**
     * 处理消费异常
     **/
    private void handlingException(OutboxEvent event, Throwable e) {
        String errorMessage;
        e.printStackTrace();
        if (e instanceof NullPointerException) {
            errorMessage = e.getClass().getName() + " ：" + Tools.getFirstStackTrace(e);
        } else {
            errorMessage = Tools.getThrowable(e).getMessage();
        }
        execute(() -> {
            event.incTimes();
            event.setState(1);
            event.setError(errorMessage);
            event.setTimestamp(LocalDateTime.now().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
            outboxEventRepository.saveAndFlush(event);
        });
        if (event.getTimes() == 1) {
            sendExceptionMessage(event);
        }
        throw new RuntimeException(errorMessage);
    }

    private void handlingNextEvent(OutboxEvent event) {
        sendAynEvent("event:outbox.message.next", MapUtil.mapper(
                "subscriber", event.getSubscriber(),
                "aggregateType", event.getAggregateType(),
                "aggregateId", event.getAggregateId()
        ));
    }

    @Listener(topic = "event:outbox.message.next")
    public void handlingNextEvent(SimpleRequest request) {
        Map row = request.getMap();
        Specification<OutboxEvent> specification = JpaDsl.toCriteria(row);
        PageRequest pageRequest = PageRequest.of(0, 1, new Sort(Sort.Direction.ASC, "id"));
        OutboxEvent outboxEvent = outboxEventRepository.findAll(specification, pageRequest).getContent().stream().findAny().orElse(null);
        if (outboxEvent == null) return;
        invoke("outbox.message.consume", MapUtil.mapper("id", outboxEvent.getId(), "timestamp", outboxEvent.getTimestamp()));
    }

    private OutboxSubscriber getOutboxSubscriber(OutboxEvent event) {
        OutboxSubscriber outboxSubscriber = outboxSubscriberManager.getOutboxSubscriber(event.getSubscriber(), event.getEventType(), event.getAggregateType());
        if (outboxSubscriber == null) {
            outboxSubscriber = outboxSubscriberManager.getOutboxSubscriber(event.getSubscriber(), event.getEventType(), "*");
        }
        if (outboxSubscriber == null) {
            throw new RuntimeException("订阅者不存在！");
        }
        return outboxSubscriber;
    }

    private Object getPayload(OutboxEvent event) {
        Object payload;
        try {
            Class clazz = Class.forName(event.getAggregateType());
            SimpleRequest sr = new SimpleRequest();
            sr.setData(event.getPayload());
            payload = sr.getO(clazz);
            if (payload instanceof BaseModel && StringUtils.hasText(event.getSessionId())) {
                ((BaseModel) payload).setSessionId(event.getSessionId());
            }
        } catch (ClassNotFoundException e) {
            payload = event.getPayload();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return payload;
    }

    /**
     * 发送异常消息
     */
    private void sendExceptionMessage(OutboxEvent event) {
        try {
            EventUtil.sendAynEvent(this, "event:invoke.exception.intercept",
                    MapUtil.mapper("topic", event.get_subscriber().getId(),
                            "description", event.get_subscriber().getName(),
                            "type", "OutBox",
                            "request", MapUtil.mapped(event),
                            "response", event.getError()));
        } catch (Exception e) {
        }
    }

    /**
     * 处理遗漏消息
     **/
    public void handlingMissingMessages(SimpleRequest request) {
        int size = 100;

        /**处理从未消费的消息**/
        Map row = request.getMap();
        Date createDate = DateUtil.add(new Date(), Calendar.YEAR, -2);
        int expire = row.containsKey("expire") ? Integer.valueOf(String.valueOf(row.get("expire"))) : 10; /**单位：分钟**/
        Long maxTimestamp = DateUtil.add(new Date(), Calendar.MINUTE, -1 * expire).getTime();
        while (true) {
            List<OutboxEvent> outboxEvents = outboxEventRepository.findByCreateDateGreaterThanAndTimestampLessThanEqualAndTimesAndHandledOrderByCreateDateAsc(createDate, maxTimestamp, 0, 0, PageRequest.of(0, size));
            createDate = pushMessages(outboxEvents);
            if (outboxEvents.size() < size) {
                break;
            }
        }

        /**处理消费中丢失的消息**/
        expire = 60; /**单位：分钟**/
        createDate = DateUtil.add(new Date(), Calendar.YEAR, -2);
        Date updateDate = DateUtil.add(new Date(), Calendar.MINUTE, -1 * expire);
        while (true) {
            List<OutboxEvent> outboxEvents = outboxEventRepository.findByCreateDateGreaterThanAndUpdateDateLessThanAndTimesLessThanOrderByCreateDateAsc(createDate, updateDate, 12, PageRequest.of(0, size));
            createDate = pushMessages(outboxEvents);
            if (outboxEvents.size() < size) {
                break;
            }
        }
    }

    private Date pushMessages(List<OutboxEvent> outboxEvents) {
        outboxEvents.stream().forEach(outboxEvent -> {
            outboxEvent.setHandled(1);
            outboxEventRepository.save(outboxEvent);
            pushMQ(outboxEvent);
        });
        return outboxEvents.isEmpty() ? null : outboxEvents.get(outboxEvents.size() - 1).getCreateDate();
    }

    public void execute(Callback callback) {
        transactionTemplate.execute(status -> {
            try {
                callback.execute();
                return null;
            } catch (Exception e) {
                status.isRollbackOnly();
                throw e;
            }
        });
    }

    @FunctionalInterface
    public interface Callback {
        @Nullable
        void execute();
    }

    public OutboxEvent saveOutboxEvent(SimpleRequest request) {
        OutboxEvent event = request.getO(OutboxEvent.class);
        if (null != event.getSubscriber() && event.getSubscriber().contains("{")) {
            event.setSubscriber((String) JSONObject.parseObject(event.getSubscriber()).get("id"));
        }
        if (null != event.getAggregateType() && event.getAggregateType().contains("{")) {
            event.setAggregateType((String) JSONObject.parseObject(event.getAggregateType()).get("id"));
        }
        event.setTimestamp(LocalDateTime.now().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());
        return outboxEventRepository.save(event);
    }

    public String delete(SimpleRequest request) {
        outboxEventRepository.delete(request.getO(OutboxEvent.class).getId());
        return "删除成功";
    }

    /**
     * 重新发送消息（慎用）
     */
    public void resend(SimpleRequest request) {
        String classSimpleName = request.get("businessType");
        Class clazz = JpaUtil.getDomainClass(classSimpleName);
        if (clazz == null) {
            throw new RuntimeException("单据类不存在！");
        }
        CustomRepository repository = JpaUtil.getCustomRepository(clazz.getName());
        //Serializable id = request.get("id");
        Serializable id = JpaUtil.getId(clazz.getName(), request.get("id"));
        Object entity = repository.findOne(id);
        if (entity == null) {
            throw new RuntimeException("实例不存在！");
        }
        List<OutboxSubscriber> subscribers = getOutboxSubscribers(entity, "S");
        if (subscribers == null || subscribers.isEmpty()) {
            return;
        }
        String subscriberId = request.get("subscriberId");
        subscribers = subscribers.stream().filter(subscriber -> subscriberId.equals(subscriber.getId())).collect(Collectors.toList());
        if (subscribers == null || subscribers.isEmpty()) {
            throw new RuntimeException("订阅者不存在！");
        }
        publish(entity, subscribers);
    }

    /**
     * 批处理消息
     **/
    public void batchProcessing(SimpleRequest request) {
        String subscriber = request.get("subscriber");
        int size = 100;
        int expire = request.get("expire",10); /**单位：分钟**/
        Long maxTimestamp = DateUtil.add(new Date(), Calendar.MINUTE, -1 * expire).getTime();
        Date createDate = DateUtil.add(new Date(), Calendar.YEAR, -2);
        while (true) {
            List<OutboxEvent> outboxEvents = outboxEventRepository.findAll(JpaDsl.toCriteria(
                    "subscriber", "EQ", subscriber,
                    "createDate", "GT", createDate,
                    "timestamp", "LTE", maxTimestamp),
                    PageRequest.of(0, size, new Sort(Sort.Direction.ASC, "id"))).getContent();
            outboxEvents.stream().forEach(outboxEvent -> {
                try {
                    execute(() -> {
                        invoke("outbox.message.consume", MapUtil.mapper("id", outboxEvent.getId(), "timestamp", outboxEvent.getTimestamp()));
                    });
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
            if (outboxEvents.size() < size) {
                break;
            }
            createDate = outboxEvents.get(outboxEvents.size() - 1).getCreateDate();
        }
    }
    /**
     * 根据单号批量消费
     **/
    public void batchProcessingByAggregateId(SimpleRequest request) {
        String id = request.get("aggregateId");
        int size = 100;
        while (true) {
            List<OutboxEvent> outboxEvents = outboxEventRepository.findAll(JpaDsl.toCriteria(
                    "aggregateId", "EQ", id),
                    PageRequest.of(0, size, new Sort(Sort.Direction.ASC, "id"))).getContent();
            outboxEvents.stream().forEach(outboxEvent -> {
                try {
                    execute(() -> {
                        invoke("outbox.message.consume", MapUtil.mapper("id", outboxEvent.getId(), "timestamp", outboxEvent.getTimestamp()));
                    });
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
            if (outboxEvents.size() < size) {
                break;
            }
        }
    }

    /**
     * 批处理消息 - 根据单号处理
     **/
    public void accordAggregateBbatchProcessing(SimpleRequest request) {
        JSONObject getCondi = request.getJO();
        List<OutboxEvent> existsEventBacks = outboxEventRepository.existsEventBack(getCondi.getString("aggregateId"), getCondi.getString("outboxSubscriberId"));
        existsEventBacks.forEach(OutboxEvent -> outboxEventRepository.delete(OutboxEvent));

        List<OutboxEvent> outboxEvents = outboxEventRepository.accordingAggregateIdQuery(getCondi.getString("aggregateId"), getCondi.getString("outboxSubscriberId"));
        outboxEvents.stream().anyMatch(outboxEvent -> {
            try {
                execute(() -> {
                    invoke("outbox.message.consume", MapUtil.mapper("id", outboxEvent.getId(), "timestamp", outboxEvent.getTimestamp()));
                });
            } catch (Exception e) {
                e.printStackTrace();
//                outboxEventRepository.delete(outboxEvent);
                if (e.getMessage().indexOf("请不要重复提交！") > 0) outboxEventRepository.delete(outboxEvent);
                if (e.getMessage().indexOf("库存") > 0 || e.getMessage().indexOf("封库") > 0 || e.getMessage().indexOf("批次") > 0) return false;
                return true; // 批量补货 - 聚合根ID 相同，内容不同，除以上错误,  出错无消费下去的必要，解决错误后继续消费
            }
            return false;
        });
    }

    // 财务审核事务发件箱堵塞, 仅限采购使用
    public void blockUse(SimpleRequest request) {
        if (outboxEventRepository.findAll().size() < 20000) return;
        List condi = new ArrayList();
        condi.add("YH");
        condi.add("CG");

        condi.forEach(aggregateId -> {
            List<OutboxEvent> outboxEvents = outboxEventRepository.accordingAggregateIdQuery(aggregateId.toString());
            outboxEvents.stream().forEach(outboxEvent -> {
                try {
                    execute(() -> {
                        invoke("outbox.message.consume", MapUtil.mapper("id", outboxEvent.getId(), "timestamp", outboxEvent.getTimestamp()));
                    });
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        });

    }
}
